package org.fastsyncer.manager.framework;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.fastsyncer.common.constant.CommonConstant;
import org.fastsyncer.common.constant.MappingConstant;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.common.util.ApplicationUtil;
import org.fastsyncer.manager.clean.DataCleanFactory;
import org.fastsyncer.manager.listener.TaskHandleListener;
import org.fastsyncer.manager.store.DataCenter;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * ManagerFactory:
 * <p>
 * 执行流程如下：1.分析任务配置 2.持久化任务信息 3.计算拆分任务 4.指派任务给worker 5.接受worker执行结果 6.刷新最新进度
 * </p>
 * 
 * @author YangPei
 * @date 2017年5月13日 下午8:06:44
 * @version 1.0.0
 */
public class ManagerFactory implements Manager, TaskHandleListener {

	private static final Logger logger = LoggerFactory.getLogger(ManagerFactory.class);

	private static ManagerFactory instance = new ManagerFactory();

	private ManagerFactory() {
	}

	public static ManagerFactory getInstance() {
		return instance;
	}

	// 支持同时处理数据同步任务并发数
	private final Integer MAX_SYNC = Integer.parseInt(ApplicationUtil.getKey(CommonConstant.MANAGER_SYNC_NUMBER));

	// 支持同时处理数据迁移任务并发数
	private final Integer MAX_MIGRATE = Integer.parseInt(ApplicationUtil.getKey(CommonConstant.MANAGER_MIGRATE_NUMBER));

	// 待执行的数据同步任务
	private final BlockingQueue<JSONObject> SYNC_QUENE = new LinkedBlockingQueue<JSONObject>(MAX_SYNC);

	// 待执行的数据迁移任务
	private final ConcurrentLinkedQueue<String> MIGRATE_QUEUE = new ConcurrentLinkedQueue<String>();

	// 正在执行的数据迁移任务
	private final ConcurrentHashMap<String, String> RUNNING = new ConcurrentHashMap<String, String>();

	// 当前正在运行同步任务数
	private Integer run = 0;

	@Override
	public void handle(MappingTask mappingTask) throws Exception {
		// 1.分析任务配置
		if (mappingTask == null) {
		    logger.error("The task configuration can not be empty!");
			return;
		}
		logger.info("The task is accepted.");

		// 2.计算总进度
		TaskDistributor.getInstance().countProcess(mappingTask);

		// 3.持久化任务
		boolean store = DataCenter.getInstance().storeMapping(mappingTask.getId(), mappingTask);
		if (!store) {
		    logger.error("The persistent task failed!");
			return;
		}
		logger.info("The persistence task is complete.");

		// 4.添加任务到队列中
		this.addTask(mappingTask.getId());
		logger.info("Added task to the queue.\r");
	}

	@Override
	public void handleSync(JSONObject t) throws Exception {
		// 这里是数据增量同步处理，这里会接收增删改操作的数据。
		// 其中，针对DB操作，我们可以接收到同步前和同步后的变化，
		// 暂时直接同步。(这里会在1.0.0版本之后实现审计功能。)
		String taskId = t.getString("taskId");
		// 1.获取对应的映射
		MappingTask mappingTask = (MappingTask) DataCenter.getInstance().getMapping(taskId);
		if (mappingTask == null) {
			throw new Exception("驱动配置:" + taskId + "未找到.");
		}
		// 2.检查驱动是否被禁用
		if (!mappingTask.isEnable()) {
			throw new Exception("驱动配置:" + taskId + "已禁用.");
		}
        // 3.检查驱动是否被启动
        if (!mappingTask.isRun()) {
            throw new Exception("驱动:" + taskId + "未启动.");
        }
		// 4.检查驱动是否为数据同步模式
		if (!mappingTask.getModel().equals(MappingConstant.SYNC_INCREMENT)) {
			throw new Exception("驱动配置必须为数据同步模式.");
		}
		// 5.过滤数据,如果没有一条满足同步条件,直接拒绝
		DataCleanFactory.getInstance().clean(mappingTask, t);
		if(t.getJSONArray("msg").length() < 1){
		    return;
        }
		// 6.触发调度
		if (SYNC_QUENE.isEmpty()) {
			// 添加数据同步任务
			addSyncTask(t);
			// 触发数据同步调度
			new TouchSyncHandle().start();
			return;
		}
		// 添加数据同步任务
		addSyncTask(t);
	}

	/**
	 * 添加执行数据同步任务
	 * 
	 * @throws Exception
	 */
	private void addSyncTask(JSONObject t) throws Exception {
		// 排号
		boolean offer = SYNC_QUENE.offer(t);
		// 检查是否添加成功，如果失败，直接抛出异常
		if(!offer){
			throw new Exception("系统繁忙,当前并发上限为"+MAX_SYNC+",请稍后再试！");
		}
	}
	
	/**
	 * 添加执行数据迁移任务
	 * @Title: addTask 
	 * @Description: 添加执行数据迁移任务
	 * @param driverId
	 * @throws Exception
	 * @return: void
	 */
	private synchronized void addTask(String driverId) throws Exception {
		// 检查执行任务状态
		validateTask(driverId);

		if (MIGRATE_QUEUE.isEmpty()) {
			// 排号
			MIGRATE_QUEUE.offer(driverId);
			// 触发数据迁移调度
			new TouchMigrateHandle().start();
		} else {
			// 排号
			MIGRATE_QUEUE.offer(driverId);
		}
	}

	/**
	 * 检查执行任务状态
	 * @Title: validateTask 
	 * @Description: 检查执行任务状态
	 * @param driverId
	 * @throws Exception
	 * @return: void
	 */
	public void validateTask(String driverId) throws Exception {
		// 检查是否已添加该执行任务
		if (MIGRATE_QUEUE.contains(driverId)) {
			throw new IllegalArgumentException("已启动该任务！");
		}

		// 检查是否已在执行
		if (RUNNING.get(driverId) != null) {
			throw new IllegalArgumentException("正在运行该任务！");
		}
	}

	/**
	 * 触发数据迁移调度
	 */
	class TouchMigrateHandle extends Thread {

		TouchMigrateHandle() {
			setName("fastsyncer touch migration thread");
		}
		
		@Override
		public void run() {
			while (!MIGRATE_QUEUE.isEmpty()) {
				try {
					// 强占同步队列
					occupySyncTask();
					// 线程处理同步任务
					String task = MIGRATE_QUEUE.poll();
					new TaskHandle(instance, task).start();
					RUNNING.put(task, task);
				} catch (Exception e) {
				    logger.error("An exception when triggering data migration scheduling:" + e.getMessage(), e.getClass());
				}
				try {
					// 避免读取任务太快，控制一下速度
					Thread.sleep(300);
				} catch (InterruptedException e) {
				    logger.error(e.getMessage());
				}
			}
		}

	}

	/**
	 * 触发数据同步调度
	 */
	class TouchSyncHandle extends Thread {
		
		TouchSyncHandle() {
			setName("fastsyncer touch synchronization thread");
		}
		
		@Override
		public void run() {
			while (!SYNC_QUENE.isEmpty()) {
				try {
					// 1.获取第一个元素
					JSONObject t = SYNC_QUENE.peek();
					if(t==null){
						break;
					}
					// 2.数据同步:路由，将数据源的数据分别同步至目标源
					RouterHandle.getInstance().router(t);
					// 3.移除第一个元素
					SYNC_QUENE.poll();
				} catch (Exception e) {
				    logger.error("An exception is triggered when data synchronization is triggered:" + e.getMessage(), e.getClass());
				}
			}
		}

	}

	@Override
	public void callback(String driverId) {
		// 移除运行任务
		RUNNING.remove(driverId);
		// 释放强占列队
		releaseSyncTask();
	}

	/**
	 * 强占同步队列
	 */
	private synchronized void occupySyncTask() throws Exception {
		if (run + 1 > MAX_MIGRATE) {
			throw new Exception("The system is busy, please try again later! Is currently running:" + run);
		}
		run += 1;
		logger.info("Occupy the queue >> available:" + (MAX_MIGRATE - run));
	}

	/**
	 * 释放强占队列
	 */
	private synchronized void releaseSyncTask() {
		run -= 1;
		logger.info("Release the queue >> available:" + (MAX_MIGRATE - run));
	}

	/**
	 * 获取正在排队的同步任务
	 * 
	 * @return
	 */
	public ConcurrentLinkedQueue<String> getQueue() {
		return MIGRATE_QUEUE;
	}

	/**
	 * 获取正在运行的同步任务
	 * 
	 * @return
	 */
	public ConcurrentHashMap<String, String> getRunning() {
		return RUNNING;
	}

	/**
	 * 获取正在运行的同步任务
	 * 
	 * @return
	 */
	public String getRunning(String taskName) {
		return RUNNING.get(taskName);
	}

	/**
	 * 移除正在运行的同步任务
	 * 
	 * @return
	 */
	public boolean removeRunning(String taskName) {
		RUNNING.remove(taskName);
		return true;
	}

	/**
	 * 移除正在排队的同步任务
	 * 
	 * @return
	 */
	public boolean removeQueue(String taskName) {
		MIGRATE_QUEUE.remove(taskName);
		return true;
	}

}
